#include "svr_skeleton.h"

static struct
{/*{{{*/
    unsigned long thrd_id;
    EP ep;
    EP_SLOT agent_slot;
    EP_SLOT *worker_slot;
}/*}}}*/
listener_ctl;

static int _conn(struct ep_active_slot *active_slot)
{/*{{{*/
#define _MAX_NUM    9999
    SOCKET new_sock=active_slot->_u._conn.sock;
    int worker_idx;
    struct _worker_info *worker;
    int idle_num;
    int idle[cnfg_info.start];
    int min_busy_client_num;
    int min_busy_idx;
    int client_num;
    char msg[50];
    int msg_size;
    int res;

    pthread_mutex_lock(&stat_info._worker_.lock);
    idle_num=0;
    min_busy_client_num=_MAX_NUM;
    min_busy_idx=-1;
    for(worker_idx=0; worker_idx<cnfg_info.start; ++worker_idx)
    {
        worker=&stat_info._worker_.worker[worker_idx];
        if(worker->client_num == 0)
        {   idle[idle_num++]=worker_idx;   }
        else
        {
            if(min_busy_client_num > worker->client_num)
            {
                min_busy_client_num=worker->client_num;
                min_busy_idx=worker_idx;
            }
        }
    }
    if(idle_num > 0) // 选取无连接的worker
    {
        worker_idx=idle[random_int(0, idle_num-1, NULL)];
        errlog("%s: choose idle worker(%d)",
                __func__, worker_idx);
    }
    else if(min_busy_client_num < _MAX_NUM) // 选取连接最少的worker
    {
        worker_idx=min_busy_idx;
    }
    else
    {
        errlog("%s: worker full, BAD", __func__);
        pthread_mutex_unlock(&stat_info._worker_.lock);
        sock_close(new_sock);
        return(-1);
    }
    client_num=++stat_info._worker_.worker[worker_idx].client_num;
    pthread_mutex_unlock(&stat_info._worker_.lock);

    CLR_BUF(msg);
    para_put_item(msg, "NEWCONN");
    para_put_item_int(msg, new_sock);
    msg_size=strlen(msg);
    if((res=ep_write(listener_ctl.ep,
                    listener_ctl.worker_slot[worker_idx],
                    msg, &msg_size)) != 0)
    {
        errlog("%s %s: listener ep_write worker(%d) fail, res:%d, "
                "abort, impossible",
                __FILE__, __func__, worker_idx, res);
        sock_close(new_sock);
        return(-1);
    }

    errlog("%s %s: listener send new slot to worker(%d) client_num:%d",
            __FILE__, __func__, worker_idx, client_num);
    return(0);
}/*}}}*/

static void _listener()
{/*{{{*/
#define CHECK_OPERATE_INFO_INTERVAL       1
    unsigned int quit=IS_FALSE;
    struct ep_active_slot *active_list;
    time_t last_operate=0;
    time_t now;
    int res;
    
    while(!quit)
    {
        if((res=ep_poll(listener_ctl.ep, &active_list)) != 0)
        {
            errlog("%s %s: ep_poll fail, ret:%d, impossible",
                    __FILE__, __func__, res);
            continue;
        }
        while(active_list)
        {
            switch(active_list->type)
            {
                case EP_SLOT_CONN:
                    _conn(active_list);
                    break;
                case EP_SLOT_ABORT:
                    errlog("%s: revoke slot(sock:%d,idx:%d) error(%d), "
                            "impossible",
                            __func__, active_list->_u._abort.sock,
                            active_list->slot_idx,
                            active_list->_u._abort.error);
                    break;
                default:
                    errlog("%s: listener poll type(%s) invalid, "
                            "impossible",
                            __func__, active_list->type);
                    break;
            }
            EP_ACTIVE_LIST_NEXT(active_list);
        }

        now=curr_timestamp(NULL);
        if(now - last_operate > CHECK_OPERATE_INFO_INTERVAL)
        {
            pthread_mutex_lock(&operate_info._listener_.lock); //lock
            if((quit=operate_info._listener_.quit))
            {
                operate_info._listener_.state=THRD_SHUTDOWN;
            }
            pthread_mutex_unlock(&operate_info._listener_.lock); //unlock

            last_operate=now;
        }
    }
    errlog("%s: listener shutdown", __func__);
    DEEP_SLEEP;
}/*}}}*/

int listener_start(SOCKET *sock_pair)
{/*{{{*/
    char addr[LONG_FILE_NM_SIZE+1];
    SOCKET sock;
    int worker_idx;
    int res;

    pthread_mutex_init(&operate_info._listener_.lock, NULL);
    operate_info._listener_.state=THRD_RUNNING;
    operate_info._listener_.quit=IS_FALSE;

    if((res=ep_create(&listener_ctl.ep)))
    {
        fprintf(stderr,
                "%s: listener(ep_create) fail, errcode:%d\n",
                SERVER_NAME, res);
        return(-1);
    }

    sprintf(addr, "svr_%s_agent_%d",
            SERVER_NAME, cnfg_info.ipckey);
    if((sock=sock_local_listen(addr)) == INVALID_SOCKET)
    {
        fprintf(stderr, "agent: open server port fail\n");
        return(-1);
    }
    if((res=ep_slot_register(listener_ctl.ep, sock, IS_TRUE,
                    &listener_ctl.agent_slot)))
    {
        fprintf(stderr,
                "%s: listener(ep_slot_register) server port fail, "
                "errcode:%d\n",
                SERVER_NAME, res);
        return(-1);
    }
    fprintf(stderr, "%s: open agent unix domain(%s)\n", SERVER_NAME, addr);

    ALLOC_MULTI_DOERR(listener_ctl.worker_slot, EP_SLOT,
            cnfg_info.start,
            fprintf(stderr, "%s: alloc listener_ctl.worker_slot fail\n",
                SERVER_NAME);
            return(-1));
    for(worker_idx=0; worker_idx<cnfg_info.start; ++worker_idx)
    {
        if((res=ep_slot_register(listener_ctl.ep,
                        sock_pair[worker_idx*2], IS_FALSE,
                        &listener_ctl.worker_slot[worker_idx])))
        {
            fprintf(stderr,
                    "%s: listener(ep_register) worker(%d) fail, "
                    "errcode:%d\n",
                    SERVER_NAME, worker_idx, res);
            return(-1);
        }
        ep_slot_set_wrap(listener_ctl.worker_slot[worker_idx]);
    }

    /*--- thread listener ---*/
    if(thrd_create((THRD_FUNC_T)_listener, (void *)NULL, THRD_DETACH,
                &listener_ctl.thrd_id))
    {
        fprintf(stderr, "%s: listener thrd create fail\n", SERVER_NAME);
        return(-1);
    }

    return(0);
}/*}}}*/
